package com.wk.springbootlearn.service.impl;

import com.wk.springbootlearn.service.HDFSService;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.springframework.stereotype.Service;

import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

@Service
public class HDFSServiceImpl implements HDFSService {

    private final String hdfsURI = "hdfs://192.168.197.128:9000";
    private final String nodeUser = "root";

    // 获取HDFS文件系统对象
    public FileSystem getFileSystem() {
        FileSystem fs = null;
        try{
           fs = FileSystem.get(URI.create(hdfsURI),new Configuration(), nodeUser);
        }catch (Exception e) {
            System.out.println("连接hdfs失败：" + e.getMessage());
        }
        return fs;
    }

    // 创建文件夹
    @Override
    public Boolean mkdir(String dirs) {
        try{
            return this.getFileSystem().mkdirs(new Path(dirs));
        }catch (Exception e) {
            throw new RuntimeException("hdfs创建文件夹失败：" + e.getMessage());
        }
    }

    // 创建文件
    @Override
    public Boolean upload(String dirs, String fileName, InputStream input) {
        FSDataOutputStream fos = null;
        try {
            boolean exists = this.getFileSystem().exists(new Path(dirs));
            if (!exists) throw new RuntimeException("路径不存在：" + dirs);
            if (dirs.equals("/")) dirs = "";
            fos = this.getFileSystem().create(new Path(dirs +"/"+ fileName));
            byte[] buffer = new byte[1024];
            int len = 0;
            while ((len = input.read(buffer)) != -1) {
                fos.write(buffer, 0, len);
            }
            IOUtils.closeStream(fos);
            return true;
        } catch (Exception e) {
            throw new RuntimeException("hdfs创建文件失败：" + e.getMessage());
        }
    }

    // 删除文件夹或文件
    @Override
    public Boolean delete(String fileName) {
        try{
            boolean exists = this.getFileSystem().exists(new Path(fileName));
            if (!exists) throw new RuntimeException("路径不存在：" + fileName);
            return this.getFileSystem().delete(new Path(fileName), true);
        }catch (Exception e) {
            throw new RuntimeException("hdfs删除文件夹或文件失败：" + e.getMessage());
        }
    }

    // 重命名文件夹或文件
    @Override
    public Boolean rename(String oldStr, String newStr) {
        try{
            boolean exists = this.getFileSystem().exists(new Path(oldStr));
            if (!exists) throw new RuntimeException("路径不存在：" + oldStr);
            return this.getFileSystem().rename(new Path(oldStr), new Path(newStr));
        }catch (Exception e) {
            throw new RuntimeException("hdfs重命名文件夹或文件失败：" + e.getMessage());
        }
    }

    // 文件是否存在
    @Override
    public Boolean exists(String fileName) {
        try {
            return this.getFileSystem().exists(new Path(fileName));
        }catch (Exception e) {
            throw new RuntimeException("hdfs文件是否存在失败：" + e.getMessage());
        }
    }

    // 下载
    @Override
    public InputStream download(String fileName) {
        try {
            boolean exists = this.getFileSystem().exists(new Path(fileName));
            if (!exists) throw new RuntimeException("路径不存在：" + fileName);
            return this.getFileSystem().open(new Path(fileName));
        }catch (Exception e) {
            throw new RuntimeException("hdfs下载失败：" + e.getMessage());
        }
    }

    // 遍历指定路径下的文件
    @Override
    public List<String> List(String path) {
        List<String> list = new ArrayList<>();
        try{
            FileStatus[] fileStatuses = this.getFileSystem().listStatus(new Path(path));
            for (FileStatus fileStatus : fileStatuses) {
                list.add(fileStatus.getPath().getName());
            }
        }catch (Exception e) {
            throw new RuntimeException("hdfs遍历指定路径下的文件失败：" + e.getMessage());
        }
        return list;
    }
}
